02 路由与JSON响应
上一篇中使用@app.route("/chat")定义了一个基础接口。实际的 Agent API 通常包含多个接口,例如查询对话历史、管理会话、上传文件、获取 Agent 状态等。
本篇介绍 Flask 路由系统和 JSON 响应处理方式,用于构建结构清晰的 Agent API。
一、路由基础
路由是URL 路径和处理函数之间的映射关系。客户端访问某个 URL 时,Flask 会根据路由规则找到对应的函数并执行。
@app.route("/chat")
def chat():
return "chat接口"访问/chat时,Flask 会执行chat()函数。
1.1 多个路由
一个应用可以定义多个路由,不同 URL 对应不同功能:
@app.route("/")
def index():
return "首页"
@app.route("/chat")
def chat():
return "对话接口"
@app.route("/health")
def health():
return "健康检查"1.2 路由末尾的斜杠
以下两个路由的行为不同:
@app.route("/projects")
def projects():
return "没有斜杠"
@app.route("/about/")
def about():
return "有斜杠"它们的访问规则不同:
| 写法 | 访问/about | 访问/about/ |
|---|---|---|
@app.route("/about") | 正常响应 | 404 |
@app.route("/about/") | 自动重定向到/about/ | 正常响应 |
建议在项目中统一一种风格,避免同类接口出现不一致的路径规则。API 接口通常不加末尾斜杠。
二、动态路由
有些 URL 会包含变量,例如通过地址查询某个会话的历史:
@app.route("/session/<session_id>")
def get_session(session_id):
# session_id会自动从URL中提取出来
return {"session_id": session_id}访问/session/abc123时,session_id的值为"abc123"。Flask 会提取 URL 中的变量片段,并作为参数传给视图函数。
2.1 类型转换器
默认情况下,URL 变量都是字符串。可以通过转换器指定变量类型:
@app.route("/user/<username>")
def show_user(username):
# username是字符串
return {"username": username}
@app.route("/post/<int:post_id>")
def show_post(post_id):
# post_id是整数
return {"post_id": post_id}
@app.route("/price/<float:amount>")
def show_price(amount):
# amount是浮点数
return {"amount": amount}内置的转换器:
| 转换器 | 说明 | 示例 |
|---|---|---|
string | 字符串(默认),不含斜杠 | /user/john |
int | 正整数 | /post/42 |
float | 正浮点数 | /price/9.99 |
path | 字符串,可以含斜杠 | /file/a/b/c.txt |
uuid | UUID字符串 | /task/550e8400-e29b-41d4-a716-446655440000 |
2.2 Agent场景示例
动态路由在 Agent API 中较常见:
@app.route("/session/<session_id>/history")
def get_history(session_id):
"""获取某个会话的对话历史"""
# 后面会从数据库或记忆中查询
return {"session_id": session_id, "messages": []}
@app.route("/agent/<agent_name>/invoke", methods=["POST"])
def invoke_agent(agent_name):
"""调用指定的Agent"""
data = request.get_json()
return {"agent": agent_name, "result": "处理完成"}三、HTTP方法
默认情况下,路由只响应 GET 请求。Agent API 通常也会使用 POST 请求,因为用户消息、配置、上下文等数据一般通过请求体提交。
3.1 指定方法
@app.route("/chat", methods=["GET", "POST"])
def chat():
if request.method == "POST":
# 处理POST请求:接收用户消息
data = request.get_json()
return {"reply": f"收到: {data.get('message', '')}"}
else:
# 处理GET请求:返回使用说明
return {"usage": "POST /chat with {message: '...'}"}3.2 按方法拆分路由
Flask 提供了按 HTTP 方法拆分的快捷装饰器,可以避免在同一个函数中通过if/else区分请求方法:
@app.get("/health")
def health_check():
"""GET请求:健康检查"""
return {"status": "ok"}
@app.post("/chat")
def chat():
"""POST请求:发送消息"""
data = request.get_json()
return {"reply": f"收到: {data.get('message', '')}"}
@app.delete("/session/<session_id>")
def delete_session(session_id):
"""DELETE请求:删除会话"""
return {"deleted": session_id}这种写法使每个 HTTP 方法对应一个独立函数,职责更清晰。
3.3 常用HTTP方法
| 方法 | 用途 | Agent API示例 |
|---|---|---|
| GET | 查询数据 | 获取对话历史、查询Agent状态 |
| POST | 创建/提交数据 | 发送消息、创建新会话 |
| PUT | 更新数据 | 更新会话配置 |
| DELETE | 删除数据 | 删除会话 |
四、获取请求数据
Agent API 的输入可能来自请求体、URL 参数、请求头或表单。Flask 通过request对象提供这些数据的访问入口。
4.1 JSON请求体
JSON 请求体是 Agent API 中最常见的数据提交方式。
from flask import request
@app.post("/chat")
def chat():
data = request.get_json()
message = data.get("message", "")
model = data.get("model", "deepseek-v4-flash")
return {"reply": f"使用{model}回复: {message}"}客户端发送:
curl -X POST http://localhost:5000/chat \
-H "Content-Type: application/json" \
-d '{"message": "你好", "model": "deepseek-v4-flash"}'4.2 URL查询参数
URL 中问号后的参数适合用于 GET 请求:
@app.get("/history")
def history():
page = request.args.get("page", 1, type=int)
size = request.args.get("size", 10, type=int)
return {"page": page, "size": size}访问/history?page=2&size=20时,page为 2,size为 20。
request.args.get()中的type=int会尝试将字符串转换为整数;转换失败时返回默认值。相比手动调用int()并捕获异常,这种方式更适合参数解析。
4.3 表单数据
传统 HTML 表单提交的数据可以通过request.form获取:
@app.post("/login")
def login():
username = request.form.get("username")
password = request.form.get("password")
return {"username": username}Agent API 通常使用 JSON,表单数据在此类接口中使用较少。
4.4 请求头
认证 Token 等信息通常放在请求头中,而不是请求体中:
@app.post("/chat")
def chat():
token = request.headers.get("Authorization", "")
if not token.startswith("Bearer "):
return {"error": "未授权"}, 401
# 提取token
api_key = token.replace("Bearer ", "")
data = request.get_json()
return {"reply": f"收到: {data.get('message', '')}"}4.5 获取方式汇总
| 数据位置 | 获取方式 | 示例 |
|---|---|---|
| JSON请求体 | request.get_json() | {"message": "你好"} |
| URL参数 | request.args.get("key") | /chat?model=deepseek |
| 表单数据 | request.form.get("key") | username=xxx |
| 请求头 | request.headers.get("X-Token") | Authorization: Bearer xxx |
| URL变量 | 函数参数 | /session/<id> |
五、返回JSON响应
Agent API 通常返回 JSON。Flask 支持直接将字典、列表等对象转换为 JSON 响应。
5.1 直接返回字典
最基础的方式是直接return一个字典或列表,Flask 会自动转换为 JSON:
@app.get("/health")
def health():
return {"status": "ok", "version": "1.0"}Flask 会自动完成以下处理:
- 把字典序列化成JSON字符串
- 设置
Content-Type: application/json响应头 - 返回200状态码
5.2 返回列表
列表也可以直接返回:
@app.get("/models")
def list_models():
return ["deepseek-v4-flash", "deepseek-v3", "gpt-4o"]5.3 自定义状态码
默认状态码是 200。如果需要返回其他状态码,可以返回一个元组:
@app.post("/chat")
def chat():
data = request.get_json()
if not data or "message" not in data:
return {"error": "缺少message参数"}, 400 # 400 Bad Request
return {"reply": "收到"}, 201 # 201 Created元组格式为(响应体, 状态码)。
5.4 自定义响应头
如果需要增加额外响应头,可以使用三元素元组:
@app.get("/data")
def get_data():
return (
{"data": [1, 2, 3]},
200,
{"X-Request-Id": "abc123", "Cache-Control": "no-cache"},
)5.5 jsonify函数
需要更明确地构造 JSON 响应时,可以使用jsonify:
from flask import jsonify
@app.get("/user")
def get_user():
return jsonify(
username="张三",
role="admin",
)jsonify会把关键字参数转成 JSON 对象,并设置正确的 Content-Type。
5.6 返回值类型汇总
| 返回值 | Flask的处理 |
|---|---|
dict或list | 自动转JSON,200状态码 |
string | 返回HTML,200状态码 |
(dict, 状态码) | 自动转JSON,自定义状态码 |
(dict, 状态码, 响应头) | 自动转JSON,自定义状态码和响应头 |
Response对象 | 直接返回,完全自定义 |
六、重定向和错误
6.1 重定向
重定向用于将客户端引导到另一个 URL:
from flask import redirect, url_for
@app.route("/")
def index():
return redirect(url_for("health")) # 重定向到 /health
@app.route("/health")
def health():
return {"status": "ok"}url_for("health")会根据函数名health生成对应的 URL/health。相比硬编码路径,使用url_for可以在路由规则变更时减少引用处的维护成本。
6.2 主动中断请求
参数不合法时,可以主动中断请求并返回错误:
from flask import abort
@app.post("/chat")
def chat():
data = request.get_json()
if not data:
abort(400, description="请求体不能为空")
message = data.get("message")
if not message:
abort(400, description="缺少message字段")
return {"reply": f"收到: {message}"}abort(400)会立即停止当前函数并返回 400 错误。
6.3 自定义错误响应
Flask 默认错误页面是 HTML,不适合 API 调用方直接处理。可以自定义错误处理器,让错误也统一返回 JSON:
from flask import jsonify
@app.errorhandler(404)
def not_found(error):
return jsonify({"error": "接口不存在"}), 404
@app.errorhandler(500)
def internal_error(error):
return jsonify({"error": "服务器内部错误"}), 500
@app.errorhandler(400)
def bad_request(error):
return jsonify({"error": str(error.description)}), 400这样可以保证接口在正常响应和错误响应中都保持一致的数据格式。
七、完整的Agent API骨架
综合上述内容,可以得到一个结构清晰的 Agent API 骨架:
from flask import Flask, request, jsonify, abort
app = Flask(__name__)
# ---- 健康检查 ----
@app.get("/health")
def health():
return {"status": "ok"}
# ---- 对话接口 ----
@app.post("/chat")
def chat():
data = request.get_json()
if not data or "message" not in data:
abort(400, description="缺少message字段")
message = data["message"]
session_id = data.get("session_id", "default")
# 这里后面会替换成真正的Agent调用
return {
"reply": f"收到: {message}",
"session_id": session_id,
}
# ---- 会话管理 ----
@app.get("/session/<session_id>")
def get_session(session_id):
return {"session_id": session_id, "messages": []}
@app.delete("/session/<session_id>")
def delete_session(session_id):
return {"deleted": session_id}
# ---- 错误处理 ----
@app.errorhandler(404)
def not_found(error):
return jsonify({"error": "接口不存在"}), 404
@app.errorhandler(400)
def bad_request(error):
return jsonify({"error": str(error.description)}), 400
if __name__ == "__main__":
app.run(debug=True)接口列表:
| 方法 | 路径 | 功能 |
|---|---|---|
| GET | /health | 健康检查 |
| POST | /chat | 发送消息 |
| GET | /session/<id> | 查询会话 |
| DELETE | /session/<id> | 删除会话 |
这构成了 Agent API 的基础结构。后续可继续接入真实 Agent 逻辑、数据库和流式输出等能力。
八、总结
本篇主要介绍接口层的基础能力:
- 路由:用
@app.route()或@app.get()/@app.post()把URL绑定到函数 - 动态路由:
<变量名>从URL提取参数,支持类型转换 - 请求数据:
request.get_json()取JSON,request.args.get()取URL参数 - JSON响应:直接return字典,Flask自动转JSON
- 错误处理:
abort()中断请求,@app.errorhandler()自定义错误格式
下一篇将介绍 Blueprint 蓝图,用于在接口数量增加时按功能拆分模块,提升项目可维护性。